//! # 核心流处理 Trait 定义
//! 
//! 定义数据流处理的核心抽象和接口

use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use std::any::{Any, TypeId};
use std::fmt::{self, Debug};
use std::sync::Arc;
use chrono::{DateTime, Utc};
use rustcloud_bus::prelude::Message;
use crate::error::{StreamError, StreamResult};

/// 类型信息
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct TypeInfo {
    /// 类型名称
    pub name: String,
    /// 类型ID
    pub type_id: TypeId,
    /// 类型描述
    pub description: Option<String>,
}

impl TypeInfo {
    /// 创建类型信息
    pub fn new<T: 'static>() -> Self {
        Self {
            name: std::any::type_name::<T>().to_string(),
            type_id: TypeId::of::<T>(),
            description: None,
        }
    }

    /// 带描述的类型信息
    pub fn with_description<T: 'static>(description: impl Into<String>) -> Self {
        Self {
            name: std::any::type_name::<T>().to_string(),
            type_id: TypeId::of::<T>(),
            description: Some(description.into()),
        }
    }

    /// 检查类型是否匹配
    pub fn matches<T: 'static>(&self) -> bool {
        self.type_id == TypeId::of::<T>()
    }
}

/// 流偏移量
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct StreamOffset {
    /// 主题名称
    pub topic: String,
    /// 分区ID
    pub partition: Option<u32>,
    /// 偏移量
    pub offset: u64,
    /// 时间戳
    pub timestamp: DateTime<Utc>,
}

impl StreamOffset {
    /// 创建新的偏移量
    pub fn new(topic: impl Into<String>, offset: u64) -> Self {
        Self {
            topic: topic.into(),
            partition: None,
            offset,
            timestamp: Utc::now(),
        }
    }

    /// 带分区的偏移量
    pub fn with_partition(mut self, partition: u32) -> Self {
        self.partition = Some(partition);
        self
    }
}

/// 流处理器 - 核心数据处理接口
#[async_trait]
pub trait StreamProcessor<I: Send + 'static, O: Send + 'static>: Send + Sync + Debug {
    /// 处理输入数据并返回输出
    async fn process(&self, input: I) -> StreamResult<O>;

    /// 获取处理器名称
    fn name(&self) -> &str;

    /// 获取输入类型信息
    fn input_type(&self) -> TypeInfo;

    /// 获取输出类型信息
    fn output_type(&self) -> TypeInfo;

    /// 是否支持批处理
    fn supports_batch(&self) -> bool {
        false
    }

    /// 批处理方法（可选）
    async fn process_batch(&self, inputs: Vec<I>) -> StreamResult<Vec<O>> {
        let mut outputs = Vec::with_capacity(inputs.len());
        for input in inputs {
            outputs.push(self.process(input).await?);
        }
        Ok(outputs)
    }

    /// 获取处理器配置
    fn config(&self) -> Option<&dyn Any> {
        None
    }

    /// 初始化处理器
    async fn initialize(&self) -> StreamResult<()> {
        Ok(())
    }

    /// 清理处理器资源
    async fn cleanup(&self) -> StreamResult<()> {
        Ok(())
    }
}

/// 流数据源 - 数据输入接口
#[async_trait]
pub trait StreamSource: Send + Sync + Debug {
    /// 数据项类型
    type Item;

    /// 获取下一个数据项
    async fn next(&mut self) -> Option<StreamResult<Self::Item>>;

    /// 检查是否已耗尽
    fn is_exhausted(&self) -> bool;

    /// 获取源名称
    fn name(&self) -> &str;

    /// 获取源配置
    fn config(&self) -> Option<&dyn Any> {
        None
    }

    /// 启动数据源
    async fn start(&mut self) -> StreamResult<()> {
        Ok(())
    }

    /// 停止数据源
    async fn stop(&mut self) -> StreamResult<()> {
        Ok(())
    }

    /// 获取当前偏移量
    async fn current_offset(&self) -> StreamResult<Option<StreamOffset>> {
        Ok(None)
    }

    /// 重置到指定偏移量
    async fn seek_to_offset(&mut self, _offset: StreamOffset) -> StreamResult<()> {
        Err(StreamError::internal_error("源不支持偏移量重置"))
    }
}

/// 流数据汇 - 数据输出接口
#[async_trait]
pub trait StreamSink<T: Send + 'static>: Send + Sync + Debug {
    /// 发送单个数据项
    async fn send(&mut self, item: T) -> StreamResult<()>;

    /// 发送批量数据项
    async fn send_batch(&mut self, items: Vec<T>) -> StreamResult<()> {
        for item in items {
            self.send(item).await?;
        }
        Ok(())
    }

    /// 刷新缓冲区
    async fn flush(&mut self) -> StreamResult<()>;

    /// 关闭数据汇
    async fn close(&mut self) -> StreamResult<()>;

    /// 获取汇名称
    fn name(&self) -> &str;

    /// 获取汇配置
    fn config(&self) -> Option<&dyn Any> {
        None
    }

    /// 是否支持批处理
    fn supports_batch(&self) -> bool {
        false
    }

    /// 获取统计信息
    async fn stats(&self) -> StreamResult<SinkStats> {
        Ok(SinkStats::default())
    }
}

/// 流处理管道 - 管理整个处理流程
#[async_trait]
pub trait StreamPipeline: Send + Sync + Debug {
    /// 启动管道
    async fn start(&self) -> StreamResult<()>;

    /// 停止管道
    async fn stop(&self) -> StreamResult<()>;

    /// 暂停管道
    async fn pause(&self) -> StreamResult<()>;

    /// 恢复管道
    async fn resume(&self) -> StreamResult<()>;

    /// 获取管道名称
    fn name(&self) -> &str;

    /// 获取管道状态
    async fn status(&self) -> StreamResult<PipelineStatus>;

    /// 获取管道统计
    async fn stats(&self) -> StreamResult<PipelineStats>;

    /// 等待完成
    async fn wait_for_completion(&self) -> StreamResult<()>;

    /// 添加处理器
    fn add_processor(&mut self, processor: Arc<dyn StreamProcessor<Message, Message>>) -> StreamResult<()>;

    /// 移除处理器
    fn remove_processor(&mut self, name: &str) -> StreamResult<()>;

    /// 获取处理器列表
    fn processors(&self) -> Vec<String>;
}

/// 管道状态
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub enum PipelineStatus {
    /// 未初始化
    Uninitialized,
    /// 正在启动
    Starting,
    /// 运行中
    Running,
    /// 暂停中
    Paused,
    /// 正在停止
    Stopping,
    /// 已停止
    Stopped,
    /// 错误状态
    Error { message: String },
}

impl fmt::Display for PipelineStatus {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            PipelineStatus::Uninitialized => write!(f, "未初始化"),
            PipelineStatus::Starting => write!(f, "正在启动"),
            PipelineStatus::Running => write!(f, "运行中"),
            PipelineStatus::Paused => write!(f, "暂停中"),
            PipelineStatus::Stopping => write!(f, "正在停止"),
            PipelineStatus::Stopped => write!(f, "已停止"),
            PipelineStatus::Error { message } => write!(f, "错误: {}", message),
        }
    }
}

/// 管道统计信息
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct PipelineStats {
    /// 总处理消息数
    pub total_processed: u64,
    /// 成功处理数
    pub successful_processed: u64,
    /// 失败处理数
    pub failed_processed: u64,
    /// 平均处理延迟（毫秒）
    pub avg_latency_ms: f64,
    /// 吞吐量（消息/秒）
    pub throughput_per_sec: f64,
    /// 启动时间
    pub start_time: Option<DateTime<Utc>>,
    /// 最后处理时间
    pub last_processed_time: Option<DateTime<Utc>>,
    /// 错误率
    pub error_rate: f64,
}

impl PipelineStats {
    /// 创建新的统计信息
    pub fn new() -> Self {
        Self::default()
    }

    /// 记录成功处理
    pub fn record_success(&mut self, latency_ms: f64) {
        self.total_processed += 1;
        self.successful_processed += 1;
        self.last_processed_time = Some(Utc::now());
        
        // 更新平均延迟（简单移动平均）
        let total = self.total_processed as f64;
        self.avg_latency_ms = (self.avg_latency_ms * (total - 1.0) + latency_ms) / total;
        
        // 更新错误率
        self.error_rate = self.failed_processed as f64 / total;
    }

    /// 记录失败处理
    pub fn record_failure(&mut self) {
        self.total_processed += 1;
        self.failed_processed += 1;
        self.last_processed_time = Some(Utc::now());
        
        // 更新错误率
        self.error_rate = self.failed_processed as f64 / self.total_processed as f64;
    }

    /// 计算吞吐量
    pub fn calculate_throughput(&mut self) {
        if let Some(start_time) = self.start_time {
            let duration = Utc::now().signed_duration_since(start_time);
            let seconds = duration.num_seconds() as f64;
            if seconds > 0.0 {
                self.throughput_per_sec = self.total_processed as f64 / seconds;
            }
        }
    }
}

/// 汇统计信息
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct SinkStats {
    /// 发送消息数
    pub messages_sent: u64,
    /// 发送失败数
    pub send_failures: u64,
    /// 平均发送延迟
    pub avg_send_latency_ms: f64,
    /// 最后发送时间
    pub last_send_time: Option<DateTime<Utc>>,
}

/// 流上下文 - 在处理过程中传递的上下文信息
#[derive(Debug, Clone)]
pub struct StreamContext {
    /// 流名称
    pub stream_name: String,
    /// 消息ID
    pub message_id: String,
    /// 处理开始时间
    pub start_time: DateTime<Utc>,
    /// 当前偏移量
    pub current_offset: Option<StreamOffset>,
    /// 自定义属性
    pub attributes: std::collections::HashMap<String, String>,
    /// 追踪ID
    pub trace_id: Option<String>,
}

impl StreamContext {
    /// 创建新的流上下文
    pub fn new(stream_name: impl Into<String>, message_id: impl Into<String>) -> Self {
        Self {
            stream_name: stream_name.into(),
            message_id: message_id.into(),
            start_time: Utc::now(),
            current_offset: None,
            attributes: std::collections::HashMap::new(),
            trace_id: None,
        }
    }

    /// 设置偏移量
    pub fn with_offset(mut self, offset: StreamOffset) -> Self {
        self.current_offset = Some(offset);
        self
    }

    /// 设置追踪ID
    pub fn with_trace_id(mut self, trace_id: impl Into<String>) -> Self {
        self.trace_id = Some(trace_id.into());
        self
    }

    /// 添加属性
    pub fn with_attribute(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
        self.attributes.insert(key.into(), value.into());
        self
    }

    /// 获取处理耗时
    pub fn elapsed(&self) -> chrono::Duration {
        Utc::now().signed_duration_since(self.start_time)
    }

    /// 获取处理耗时（毫秒）
    pub fn elapsed_millis(&self) -> f64 {
        self.elapsed().num_milliseconds() as f64
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_type_info() {
        let type_info = TypeInfo::new::<String>();
        assert!(type_info.matches::<String>());
        assert!(!type_info.matches::<i32>());
    }

    #[test]
    fn test_stream_offset() {
        let offset = StreamOffset::new("test-topic", 100)
            .with_partition(1);
        
        assert_eq!(offset.topic, "test-topic");
        assert_eq!(offset.offset, 100);
        assert_eq!(offset.partition, Some(1));
    }

    #[test]
    fn test_pipeline_status() {
        let status = PipelineStatus::Running;
        assert_eq!(status.to_string(), "运行中");
        
        let error_status = PipelineStatus::Error { 
            message: "test error".to_string() 
        };
        assert_eq!(error_status.to_string(), "错误: test error");
    }

    #[test]
    fn test_pipeline_stats() {
        let mut stats = PipelineStats::new();
        
        stats.record_success(10.0);
        assert_eq!(stats.total_processed, 1);
        assert_eq!(stats.successful_processed, 1);
        assert_eq!(stats.avg_latency_ms, 10.0);
        assert_eq!(stats.error_rate, 0.0);
        
        stats.record_failure();
        assert_eq!(stats.total_processed, 2);
        assert_eq!(stats.failed_processed, 1);
        assert_eq!(stats.error_rate, 0.5);
    }

    #[test]
    fn test_stream_context() {
        let context = StreamContext::new("test-stream", "msg-123")
            .with_trace_id("trace-456")
            .with_attribute("key", "value");
        
        assert_eq!(context.stream_name, "test-stream");
        assert_eq!(context.message_id, "msg-123");
        assert_eq!(context.trace_id, Some("trace-456".to_string()));
        assert_eq!(context.attributes.get("key"), Some(&"value".to_string()));
    }
}